/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2017-2027 Chinazyjr,Inc.All Rights Reserved.
 */
package com.sys.midware.mongo.core.handler;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;
import com.sys.midware.mongo.core.MongoColumnAttr;
import com.sys.midware.mongo.core.MongoConnection2X;
import com.sys.midware.mongo.core.MongoUtil;
import com.sys.midware.mongo.core.index.MongoIndex;
import com.sys.midware.mongo.core.mongoOP.MongoPage;
import com.sys.midware.mongo.core.mongoOP.MongoQuery;
import com.sys.midware.mongo.core.mongoOP.MongoSort;
import com.sys.midware.mongo.core.mongoOP.MongoUpdate;
import com.sys.midware.mongo.exceptions.MongoDALException;

/**
 * 实现HBase表操作的类 <b>ClassName：</b> HBaseDBHandler <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月28日 下午2:59:30 <br/>
 * <b>@version: </b> <br/>
 */
public class MongoDBHandler2X implements IMongoDBHandler {

    private static Logger logger = LoggerFactory.getLogger(MongoDBHandler2X.class);
    
    private static int batchSize=10000;

    // define
    private String databaseName = null;

    private DB db = null;
    
    private String mongoInstanceName;

    public MongoDBHandler2X(final String mongoInstanceName, final String databaseName) throws MongoDALException {
        this.databaseName = databaseName;
        this.mongoInstanceName = mongoInstanceName;
    }
    
    public MongoDBHandler2X(final String mongoInstanceName) throws MongoDALException {
        this.mongoInstanceName = mongoInstanceName;
    }

    public List<String> getAllDataBaseNames(){
        return MongoConnection2X.getAllDataBaseName(mongoInstanceName);
    }
    
    
    @Override
    public DB openDatabase() throws MongoDALException {
        try {
            db = MongoConnection2X.getDB(mongoInstanceName, databaseName);
        } catch (Exception e) {
            logger.error("", e);
            throw new MongoDALException("fail to openDatabase." + databaseName);
        }
        return db;
    }
    
    @Override
    public DB openDatabase(String databaseName) throws MongoDALException {
        try {
            db = MongoConnection2X.getDB(mongoInstanceName, databaseName);
        } catch (Exception e) {
            logger.error("", e);
            throw new MongoDALException("fail to openDatabase." + databaseName);
        }
        return db;
    }

    @Override
    public void dropDatabase() throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
        db.dropDatabase();
    }

    @Override
    public Collection<String> getAllCollections() throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
        return db.getCollectionNames();
    }
    
    @Override
    public DBCollection getDBCollection(String tableName) throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
        return db.getCollection(tableName);
    }

    @Override
    public boolean checkTableExist(String tableName) throws MongoDALException {
        if (db == null) {
            throw new MongoDALException("DB is not open");
        }
        return db.collectionExists(tableName);
    }

    @Override
    public void performCreate(String collectionName, Map<String, MongoColumnAttr> columnMap) throws MongoDALException {
        try {
            if (db == null)
                throw new MongoDALException("DB is not open");
            // 预先处理好数据
            DBObject object = new BasicDBObject();
            String fieldName = null;
            MongoColumnAttr fieldDesc = null;
            // String STRING, INTEGER, DOUBLE
            for (Map.Entry<String, MongoColumnAttr> entity : columnMap.entrySet()) {
                fieldName = entity.getKey();
                fieldDesc = entity.getValue();
                if ("STRING".equals(fieldDesc.getColumnType().name()))
                    object.put(fieldName, "");
                else if ("INTEGER".equals(fieldDesc.getColumnType().name()))
                    object.put(fieldName, 0);
                else if ("DOUBLE".equals(fieldDesc.getColumnType().name()))
                    object.put(fieldName, false);
                else
                    throw new MongoDALException("NoSqlColumnType类型错误");
            }
//            // 开始connection
//            db.requestStart();
//            db.requestEnsureConnection();
            if (this.checkTableExist(collectionName))
                throw new MongoDALException("collection:" + collectionName + "已经存在");
            /**
             * c.ensureIndex(new BasicDBObject("dochash", 1), new
             * BasicDBObject("unique", true).append( "dropDups", true));
             */
            // create index
            DBCollection collection = db.createCollection(collectionName, object);
            DBObject indexObject = null, indexUniqueObject = null;
            for (Map.Entry<String, MongoColumnAttr> entity : columnMap.entrySet()) {

                fieldName = entity.getKey();
                fieldDesc = entity.getValue();

                if (fieldDesc.isCreateIndex()) {
                    indexObject = new BasicDBObject(fieldName, 1);
                    if (fieldDesc.isPrimaryKey() || fieldDesc.isUnique()) {
                        indexUniqueObject = new BasicDBObject("unique", true).append("dropDups", true);
                        if (indexUniqueObject != null) {
//                            collection.ensureIndex(indexObject, indexUniqueObject);
                            collection.createIndex(indexObject,indexUniqueObject);
                        }
                    } else {
//                        collection.ensureIndex(indexObject);
                        collection.createIndex(indexObject);
                    }
                }
            }

        } catch (Exception e) {
            throw new MongoDALException(e);
        } finally {
//            db.requestDone();// 结束该connection
        }
    }

    @Override
    public void performDrop(String collectionName) throws MongoDALException {
        try {
            if (db == null)
                throw new MongoDALException("DB is not open");
//            // 开始connection
//            db.requestStart();
//            db.requestEnsureConnection();
            DBCollection table = db.getCollection(collectionName);
            table.drop();
        } catch (Exception e) {
            throw new MongoDALException(e);
        } finally {
//            db.requestDone();// 结束该connection
        }
    }

    public void performAddIndex(String collectionName, DBObject index) throws MongoDALException {
        DBCollection coll =null;
        try {
            if (db == null)
                throw new MongoDALException("DB is not open");
            coll = db.getCollection(collectionName);
            if (index != null && coll != null) {
                coll.createIndex(index);
            }
        } catch (Exception e) {
            throw new MongoDALException("performAddIndex error!",e);
        }
       
    }
    
    public void performAddIndex(String collectionName, MongoIndex index) throws MongoDALException {
        DBCollection coll = null;
        try {
            if (db == null)
                throw new MongoDALException("DB is not open");
            coll = db.getCollection(collectionName);
            if (index != null && coll != null) {
                DBObject idxKeys = index.getIndexKeys();
                DBObject idxOptions = index.getIndexOptions();
                if(idxOptions!=null ){
//                    coll.ensureIndex(idxKeys,idxOptions);
                    
                    coll.createIndex(idxKeys,idxOptions); 
                }else{
                    coll.createIndex(idxKeys); 
                }
            }
        } catch (Exception e) {
            throw new MongoDALException("performAddIndex error!",e);
        }
        
    }

    @Override
    public String getDatabaseName() {
        return db==null?null:db.getName();
    }


    private void insert(DBCollection coll, Object obj) {
        DBObject dbo = MongoUtil.convertToDBObject(obj);
        coll.insert(dbo);
    }

    @Override
    public void performInsert(String collectionName, Object obj) throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");

        if (obj == null) {
            throw new IllegalArgumentException("arugument obj is not null");
        }
        try {
//            db.requestStart();
//            db.requestEnsureConnection();
            DBCollection coll = db.getCollection(collectionName);
            // single insert
            if (obj != null)
                insert(coll, obj);

        } catch (Exception e) {
            logger.error("fail to performInsert!", e);
            throw new MongoDALException("fail to performInsert!", e);
        } finally {
//            db.requestDone();
        }
    }

    @Override
    public void performBatchInsert(String collectionName, Collection<?> records) throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");

        if (records == null) {
            throw new IllegalArgumentException("arugument records is not null");
        }
        try {
//            db.requestStart();
//            db.requestEnsureConnection();
            DBCollection coll = db.getCollection(collectionName);
            // batch insert
            if (records != null && records.size() > 0) {// 此处加入验证机制
                for (Object obj : records) {
                    insert(coll, obj);
                }
            }
        } catch (Exception e) {
            logger.error("fail to performBatchInsert", e);
            throw new MongoDALException("fail to performBatchInsert", e);
        } finally {
//            db.requestDone();
        }

    }

    /**
     * 更新
     * 
     * @param collection
     * @param setFields
     * @param whereFields
     */
    private void update(DBCollection collection, MongoUpdate setFields, MongoQuery whereFields, boolean upMulti,
            boolean upsert) {
        DBObject setFieldsDBO = setFields.get();
        DBObject whereFieldsDBO = whereFields.get();
        collection.update(whereFieldsDBO, setFieldsDBO, upMulti, upsert);
    }

    @Override
    public int performUpdate(String collectionName, MongoUpdate setFields, MongoQuery whereFields, boolean upMulti,
            boolean upsert) throws MongoDALException {
        try {
//            // 开始connection
//            db.requestStart();
//            db.requestEnsureConnection();
            DBCollection table = db.getCollection(collectionName);
            update(table, setFields, whereFields, upMulti, upsert);
            return 0;
        } catch (Exception e) {
            throw new MongoDALException(e);
        } finally {
//            db.requestDone();
        }
    }

    @Override
    public int performUpdate(String collectionName, MongoUpdate setFields, MongoQuery whereFields)
            throws MongoDALException {
        try {
//            db.requestStart();
//            db.requestEnsureConnection();
            DBCollection table = db.getCollection(collectionName);
            update(table, setFields, whereFields, false, false);
            return 0;
        } catch (Exception e) {
            throw new MongoDALException(e);
        } finally {
//            db.requestDone();
        }
    }

    @Override
    public int performDelete(String collectionName, MongoQuery whereFields) throws MongoDALException {
        try {
            if (!this.checkTableExist(collectionName))
                throw new MongoDALException("collection:" + collectionName + "不存在");
            // 开启事务
//            db.requestStart();
//            db.requestEnsureConnection();
//            long beginTime = System.currentTimeMillis();
            DBCollection coll = db.getCollection(collectionName);
            DBObject whereFieldsDBO = whereFields.get();
//            logger.info("begin to performDelete,collectionName" + collectionName +",whereFieldsDBO="+whereFieldsDBO.toString());
            coll.remove(whereFieldsDBO);
//            long endtime = System.currentTimeMillis();
//            logger.info("end to performDelete,耗时：" +(endtime- beginTime) + "毫秒");
//            WriteResult wr = coll.remove(whereFieldsDBO, WriteConcern.SAFE);
//            if (null == wr) {
//                return 0;
//            }
//            return wr.getN();
            return 0;
        } catch (Exception e) {
            throw new MongoDALException("fail to performDelete!", e);

        } finally {
//            db.requestDone();
        }
    }

    /**
     * 查找（返回一个对象）
     * 
     * @param map
     * @param collection
     * @throws MongoDALException
     */
    public <T extends Object> Object find(String collectionName, MongoQuery paramObj, Class<T> bean)
            throws MongoDALException {
        if (collectionName == null || paramObj == null) {
            throw new IllegalArgumentException("arugument obj is not null");
        }
        try {
            DBObject dboResult = null;
            DBCollection coll = db.getCollection(collectionName);
            dboResult = coll.findOne(paramObj.get());
            return MongoUtil.DBObject2Bean(dboResult, bean);
        } catch (Exception e) {
            logger.error("fail to find", e);
            throw new MongoDALException("fail to find", e);
        }
    }

    private  Collection<?> query(String collectionName, Map<String, Object> columns,
            MongoQuery queryCondition, MongoPage mongopage, Class<?> beanClazz) throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
        DBCollection coll = db.getCollection(collectionName);
        DBCursor cursor = null;
        // 判断 查询条件是否为null
        DBObject qc = null;
        if (queryCondition == null) {
            qc = new BasicDBObject();
        } else {
            qc = queryCondition.get();
        }

        // 返回指定字段
        DBObject keysObj = null;
        BasicDBObject keysBasicObj = null;
        if (columns != null && !columns.isEmpty()) {
            keysObj = MongoUtil.map2DBObject(columns);
            if (keysObj != null) {
                keysBasicObj = (BasicDBObject) keysObj;
                // 依据columns返回指定的字段
                for (Map.Entry<String, Object> entity : keysBasicObj.entrySet()) {
                    entity.setValue(1);// 设置为要返回的字段
                }
            }
        }

        DBObject sc = null;
        DBObject sortCondition = mongopage.getSortObject();
        if (sortCondition == null) {
            sc = new BasicDBObject();
        } else {
            sc = (DBObject) sortCondition;
        }
        int limit = mongopage.getLimit();
        int skip = mongopage.getSkip();

        // 查询
        if (limit >= 0) {
            if (keysBasicObj != null) {
                cursor = coll.find(qc, keysBasicObj).sort(sc).skip(skip).limit(limit);
            } else {
                cursor = coll.find(qc).sort(sc).skip(skip).limit(limit);
            }
        } else {
            if (keysBasicObj != null) {
                cursor = coll.find(qc, keysBasicObj).sort(sc);
            } else {
                cursor = coll.find(qc).sort(sc);
            }
        }
        List<Object> col = new ArrayList<Object>();
        try {
            while (cursor.hasNext()) {
                DBObject object = cursor.next();
                Object beanObj = null;
                try {
                    beanObj = MongoUtil.DBObject2Bean(object, beanClazz);
                    col.add(beanObj);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        } finally {
            cursor.close();
        }
        return col;
    }

    @Override
    public Collection<?> performQuery(String collectionName, MongoQuery queryCondition, int limit, Class<?> beanClazz)
            throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
        MongoPage page = MongoPage.start();
        page.limit(limit);
        return query(collectionName, null, queryCondition, page, beanClazz);
    }

    @Override
    public Collection<?> performQuery(String collectionName, MongoQuery queryCondition, MongoSort sortCondition,
            int limit, Class<?> beanClazz) throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
        MongoPage page = MongoPage.start();
        page.sort(sortCondition);
        page.limit(limit);
        return query(collectionName, null, queryCondition, page, beanClazz);
    }

    @Override
    public Collection<?> performQuery(String collectionName, MongoQuery queryCondition, int pageNo, int pageSize,
            Class<?> beanClazz) throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
        MongoPage page = MongoPage.start();
        page.pagination(pageNo, pageSize);
        return query(collectionName, null, queryCondition, page, beanClazz);
    }

    @Override
    public Collection<?> performQuery(String collectionName, Map<String, Object> columns, MongoQuery queryCondition,
            MongoPage mongoPage, Class<?> beanClazz) throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
        return query(collectionName, columns, queryCondition, mongoPage, beanClazz);
    }

    private <T extends Object> void query(String collectionName, Map<String, Object> columns,
            MongoQuery queryCondition, MongoPage mongopage,Class<?> beanClazz, LinkedBlockingQueue<Map<String,Object>> queue) throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
        DBCollection coll = db.getCollection(collectionName);
        DBCursor cursor = null;
        // 判断 查询条件是否为null
        DBObject qc = null;
        if (queryCondition == null) {
            qc = new BasicDBObject();
        } else {
            qc = queryCondition.get();
        }

        // 返回指定字段
        DBObject keysObj = null;
        BasicDBObject keysBasicObj = null;
        if (columns != null && !columns.isEmpty()) {
            keysObj = MongoUtil.map2DBObject(columns);
            if (keysObj != null) {
                keysBasicObj = (BasicDBObject) keysObj;
                // 依据columns返回指定的字段
                for (Map.Entry<String, Object> entity : keysBasicObj.entrySet()) {
                    entity.setValue(1);// 设置为要返回的字段
                }
            }
        }

        DBObject sc = null;
        DBObject sortCondition = mongopage.getSortObject();
        if (sortCondition == null) {
            sc = new BasicDBObject();
        } else {
            sc = (DBObject) sortCondition;
        }
        int limit = mongopage.getLimit();
        int skip = mongopage.getSkip();

        // 查询
        if (limit >= 0) {
            if (keysBasicObj != null) {
                cursor = coll.find(qc, keysBasicObj).sort(sc).batchSize(batchSize).skip(skip).limit(limit);
            } else {
                cursor = coll.find(qc).sort(sc).skip(skip).batchSize(batchSize).limit(limit);
            }
        } else {
            if (keysBasicObj != null) {
                cursor = coll.find(qc, keysBasicObj).batchSize(batchSize).sort(sc);
            } else {
                cursor = coll.find(qc).batchSize(batchSize).sort(sc);
            }
        }
        try {
            while (cursor.hasNext()) {
                DBObject object = cursor.next();
                Map<String,Object> map=null;
                try {
                    map= MongoUtil.dualDBObj4Map(object);
                    if(map!=null){
                        queue.offer(map);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        } finally {
            cursor.close();
        }
    }
    @Override
    public void performQuery(String collectionName, Map<String, Object> columns, MongoQuery queryCondition,
            MongoPage mongoPage,Class<?> beanClazz, LinkedBlockingQueue<Map<String,Object>> queue) throws MongoDALException {
        if (db == null)
            throw new MongoDALException("DB is not open");
         query(collectionName, columns, queryCondition, mongoPage, beanClazz,queue);
    }
    
    
    
    @Override
    public long count(String collectionName) {
        try {
            if (!this.checkTableExist(collectionName))
                throw new Exception("collection:" + collectionName + "不存在.");

            return db.getCollection(collectionName).count();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return -1;
    }

    @Override
    public long findCountBy(String collectionName, MongoQuery queryCondition) {
        try {
            if (!this.checkTableExist(collectionName)) {
                throw new Exception("collection:" + collectionName + "不存在.");
            }
            DBObject qc = null;
            if (queryCondition == null) {
                qc = new BasicDBObject();
            } else {
                qc = queryCondition.get();
            }
            return db.getCollection(collectionName).getCount(qc);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return -1;
    }

}
